/*
	TCP 慢开始
	2023/12/20
	王威
* */

package com.ouc.tcp.test;

import com.ouc.tcp.client.TCP_Sender_ADT;
import com.ouc.tcp.client.UDT_Timer;
import com.ouc.tcp.message.TCP_PACKET;

import java.util.concurrent.LinkedBlockingQueue;

// TCP Tahoe
enum TCP_State {
    SlowStart,           // 慢开始
    CongestionAvoidance, // 拥塞避免
    FastRetransmit,      // 快重传
    FastRecovery,        // 快恢复
}

public class TCP_Sender extends TCP_Sender_ADT {

    private TCP_PACKET tcpPack;    //待发送的TCP数据报
    private int cwind = 1; // 滑动窗口大小
    private int send_base = 1; // 滑动窗口起始位置
    private int nextseqnum = 1; // 下一个可以发送位置
    private int nextResend = 1; // 下一个要重传的位置
    private int ssthresh = 16; // 慢开始门限
    private boolean resend = false;// 超时重发状态,此时阻塞由应用层的调用
    private LinkedBlockingQueue<TCP_PACKET> windows; // 滑动窗口，使用循环链表实现
    private UDT_Timer reTransTimer;
    private RetransTask reTrans; // 超时重传
    private int prevAck = 0; // 之前被ack的包
    private int ackTimes = 0; // 之前被ack的包的次数
    private volatile TCP_State state = TCP_State.SlowStart; // 标识现在TCP sender的状态
    private double cwindTmp = 0; // cwindTmp是用来进行拥塞避免使用

    /*构造函数*/
    public TCP_Sender() {
        super();    //调用超类构造函数
        super.initTCP_Sender(this);        //初始化TCP发送端
        windows = new LinkedBlockingQueue<TCP_PACKET>();
        reTransTimer = new UDT_Timer();
    }

    // 更新窗口
    private void slideWindows(int s) {
        for (int i = 0; i < s; i++) {
            windows.poll();
        }
    }

    // 没有问题
    @Override
    //可靠发送（应用层调用）：封装应用层数据，产生TCP数据报；需要修改
    // rdt_send是从上层获取到的数据
    public void rdt_send(int dataIndex, int[] appData) {
        // 等待未发送的包都全部先送出去
        // 窗口满了
        while (nextseqnum >= send_base + cwind || resend);
        //生成TCP数据报（设置序号和数据字段/校验和),注意打包的顺序
        tcpH.setTh_seq(nextseqnum);//包序号设置包的个数：
        tcpS.setData(appData); // 数据部分
        tcpPack = new TCP_PACKET(tcpH, tcpS, destinAddr); //将TCP打包成一个完整的包
        //更新带有checksum的TCP 报文头
        tcpH.setTh_sum(CheckSum.computeChkSum(tcpPack));
        tcpPack.setTcpH(tcpH); // 给TCP首部设置校验和
        // 发送该包
        udt_send(tcpPack);
        // 添加到滑动窗口要重传的数据中
        if (windows.size() < cwind) {
            // System.out.println("Add success");
            try {
                windows.offer(tcpPack.clone());
            } catch (CloneNotSupportedException e) {
                throw new RuntimeException(e);
            }
        }

        if (send_base == nextseqnum) {
            // 设置超时重传
            reTrans = new RetransTask(windows, this);
            reTransTimer.schedule(reTrans, 3000, 3000);
        }
        nextseqnum++;
    }

    @Override
    //不可靠发送：将打包好的TCP数据报通过不可靠传输信道发送；仅需修改错误标志
    public void udt_send(TCP_PACKET stcpPack) {
        //设置错误控制标志
        tcpH.setTh_eflag((byte) 7);
        //发送数据报
        client.send(stcpPack);
        // System.out.println("send Pack sequence " + stcpPack.getTcpH().getTh_seq());
        // }
    }

    // 等待ACK可能会出现的情况
    // 1、ACK是累计确认没有出错
    // 2、ACK出错，无法识别这个ACK
    // 3、超过一段时间没有收到ACK
    // 4、连续收到3个对同一个包的ACK，快重传
    @Override
    public void waitACK() {
        //循环检查ackQueue
        //循环检查确认号对列中是否有新收到的ACK
        // bug 当第一个包超时重传时，直接导致系统死锁
        // PrintWindwos();
        if (!ackQueue.isEmpty()) {
            System.out.println("current windows size " + cwind);
            int currentAck = ackQueue.poll();
            if (currentAck == prevAck) {
                ackTimes++;
                if (ackTimes == 3) {
                    // 快重传
                    TCP_PACKET pack = windows.peek();
                    System.out.println("3-ack, resend pack: " + pack.getTcpH().getTh_seq());
                    System.out.println("Start FastRetransmit");
                    udt_send(pack); // 重传
                    System.out.println("Change the previous window size " + cwind);
                    System.out.println("Change the previous ssthresh " + ssthresh);
                    cwind = Math.max(cwind / 2, 1); // 避免cwind == 0
                    ssthresh = cwind;
                    cwindTmp = cwind;
                    System.out.println("Changed window size " + cwind);
                    System.out.println("Changed ssthresh " + ssthresh);

                    // System.out.println("Start CongestionAvoidance");
                    state = TCP_State.CongestionAvoidance; // 执行拥塞避免
                }
            } else {
                // 表示当前接收的包在窗口内部
                prevAck = currentAck;
                ackTimes = 1;
            }

            if (currentAck >= send_base) {
                int slide = currentAck + 1 - send_base;
                send_base = currentAck + 1;
                slideWindows(slide);
                // 因为滑动窗口收缩导致有些ack不在窗口
                // 如果重传了更新
                // 否则忽略
                if (send_base == nextseqnum) {
                    // 证明该窗口包全部收到了
                    // System.out.println("Cancel reTrans");
                    reTrans.cancel();
                } else {
                    // 重新设置计时器
                    reTrans.cancel();
                    reTrans = new RetransTask(windows, this);
                    reTransTimer.schedule(this.reTrans, 3000, 3000);
                    // System.out.println("Restart reTrans");
                }

                // 修改更新窗口逻辑
                // System.out.println("Start change cwind");
                if (cwind >= ssthresh) {
                    state = TCP_State.CongestionAvoidance; // 拥塞避免
                    // System.out.println("Start CongestionAvoidance");
                    cwindTmp += 1.0 / cwind;
                    cwind = (int) cwindTmp;
                } else {
                    cwind++;
                    cwindTmp++;
                }
                // System.out.println("current cwind " + cwind);
            } else {
                // 收到相同的包，是延时的包
                // System.out.println("Receive delay Ack " + currentAck);
            }
        }
    }


    // 不可能会出错的
    @Override
    //接收到ACK报文：检查校验和，将确认号插入ack队列
    public void recv(TCP_PACKET recvPack) {
        // 这时候recv只能无脑接收包了，无法确定是对哪一个包进行ack，可以提前计算校验和
        if (CheckSum.computeChkSum(recvPack) == recvPack.getTcpH().getTh_sum()) {
            System.out.println("Receive ACK Number： " + recvPack.getTcpH().getTh_ack());
            ackQueue.add(recvPack.getTcpH().getTh_ack());
            //处理ACK报文
            waitACK();
        }
    }

    // TCP 慢开始
    public void setResend(boolean flag) {
        this.resend = flag;
    } // 设置重传标记位

    public void setSsthresh(int size) {
        this.ssthresh = Math.max(ssthresh, 2);
    } // 慢开始门限

    public void setTCPState(TCP_State state) {
        this.state = state;
    }

    public void setNextseqnum() {
        this.nextResend = this.send_base + 1;
    }

    public int getCwind() {
        return this.cwind;
    } // 获得窗口大小

    public void setCwind(int size) {
        this.cwind = size;
    } // 设置窗口大小

    public void setSendBase(int send_base) {
        this.send_base = send_base;
    } // 更新窗口

    public void PrintWindwos() {
        System.out.println("Current windows");
        for (TCP_PACKET pack : windows) {
            System.out.print(pack.getTcpH().getTh_seq() + " ");
        }
        System.out.println();
    }
}
